package io.polly.ldz.reactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class GenerateDemo {

    public static void main(String[] args) {

        Flux.range(1, 10)
                .publishOn(Schedulers.single())
                .map(i -> i * 2+"d")
                .handle((i, t) -> {
            t.next(i+"i");
        }).subscribe(System.out::println);
    }



    public void create() {


        //Asynchronous and Multi-threaded
        Flux<String> bridge = Flux.create(sink -> {

            List<String> myList = Arrays.asList("Hello", "World");

            MyEventProcessor.register(
                    new MyEventListener<String>() {

                        public void onDataChunk(List<String> chunk) {
                            for(String s : chunk) {
                                sink.next(s);
                            }
                        }

                        public void processComplete() {
                            sink.complete();
                        }
                    },myList);

        });

        bridge.subscribe(System.out::println);
    }



    public void generate(){
        Flux<Object> flux = Flux.generate(
//                ()->3,
                AtomicLong::new,
                (state,sink) -> {
                    Long i= state.getAndIncrement();
                    if(i<10){
                        sink.next("Hello"+i);
                    }else {
                        sink.complete();
                    }

                    return state;
                });
        flux.subscribe(System.out::println);
    }
}
